package Yanxin.atguigu.yx.app.APP_01_BaseApp;

import Yanxin.atguigu.yx.util.FlinkSourceUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;

public abstract class BaseApp_moreTopic {

    //todo 初始化flink的app，
    // 1.包括获取环境，
    // 2.设置ckp
    // 3.对外提供环境与初始流的方法handle

    public void init(int port, int p, String ckpPath_GroupId_JobName, String...topics){

        if (topics.length == 0) {
            throw new RuntimeException("传入的topic数量不对");
        }

        //设置hadoop的用户权限,大写：ctrl + shift + U
        System.setProperty("HADOOP_USER_NAME","atguigu");

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(p);

        env.enableCheckpointing(3000);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu2022/"+ ckpPath_GroupId_JobName);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置超时时间
        env.getCheckpointConfig().setCheckpointTimeout(20*1000);
        //设置ckp并发度
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //设置ckp的运行间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //程序重启后ckp清理ckp:
        // Delete externalized checkpoints on job cancellation 删除作业取消时的外部化检查点
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        //创建一个hashMap集合
        HashMap<String, DataStreamSource<String>> streams = new HashMap<>();
        for (String topic : topics) {
            DataStreamSource<String> stream = env.addSource(FlinkSourceUtil.getKafkaSource(ckpPath_GroupId_JobName, topic));
            streams.put(topic,stream);
        }

        //todo 设计一个抽象方法handle，让子类必须重写该方法，该方法主要用于完成对流内部数据处理的逻辑
        handle(env,streams);

         try {
             //执行的时候，线程名字由ckpPath_GropuId_JobName进行动态命名
             env.execute(ckpPath_GroupId_JobName);
         } catch (Exception e) {
             e.printStackTrace();
         }
    }


    //todo 让子类强行重写的方法
    protected abstract void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams);

}
